package com.earthchen.config.mqtt.in;

import com.earthchen.config.mqtt.MqttProperties;
import com.earthchen.domain.dto.MattressDTO;
import com.earthchen.service.MattressService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.io.IOException;


/**
 * MQTT消息接收处理
 *
 * @author earthchen
 * @date 2018/10/15
 **/
@Configuration
public class MqttInboundConfiguration {

    private Logger logger = LoggerFactory.getLogger(MqttInboundConfiguration.class);


    @Autowired
    private MattressService mattressService;

    @Autowired
    private MqttProperties mqttProperties;


    private ObjectMapper objectMapper = new ObjectMapper();

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        String[] inboundTopics = mqttProperties.getInbound().getTopics().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(mqttProperties.getInbound().getUrl(), mqttProperties.getInbound().getClientId(),
                        inboundTopics);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttInputChannel")
    public MessageHandler handler() {
        return message -> {
            logger.info((String) message.getPayload());
            try {
                MattressDTO mattressDTO = objectMapper.readValue((String) message.getPayload(), MattressDTO.class);
                mattressService.syncMqttToDb(mattressDTO);
            } catch (IOException e) {
                logger.error("序列化失败");
            }
        };
    }
}
